package com.ultrapower.mapper;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

import com.sematext.hbase.wd.RowKeyDistributorByHashPrefix;
import com.ultrapower.bean.ApplicationBean;
import com.ultrapower.buffer.Buffer;
import com.ultrapower.buffer.FixedBuffer;
import com.ultrapower.buffer.OffsetFixedBuffer;
import com.ultrapower.util.CommonUtils;
import com.ultrapower.util.topology.LinkDataMap;

/**
 * rowkey = caller col = callee
 *
 * @author
 */
public class MapStatisticsCallerMapper implements RowMapper<LinkDataMap> {

	private RowKeyDistributorByHashPrefix rowKeyDistributorByHashPrefix;

	public MapStatisticsCallerMapper(RowKeyDistributorByHashPrefix rowKeyDistributorByHashPrefix) {
		this.rowKeyDistributorByHashPrefix = rowKeyDistributorByHashPrefix;
	}

	@Override
	public LinkDataMap mapRow(Result result, int rowNum) throws Exception {
		if (result.isEmpty()) {
			return new LinkDataMap();
		}

		final byte[] rowKey = getOriginalKey(result.getRow());

		final Buffer row = new FixedBuffer(rowKey);
		final ApplicationBean caller = readCallerApplication(row);
		final long timestamp = CommonUtils.recoveryTimeMillis(row.readLong());

		// key is destApplicationName.
		final LinkDataMap linkDataMap = new LinkDataMap();
		for (Cell cell : result.rawCells()) {
			final Buffer buffer = new OffsetFixedBuffer(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
			final ApplicationBean callee = readCalleeApplication(buffer);

			String calleeHost = buffer.readPrefixedString();
			short histogramSlot = buffer.readShort();
			boolean isError = histogramSlot == (short) -1;
			String callerAgentId = buffer.readPrefixedString();
			long requestCount = getValueToLong(cell);

			final short slotTime = (isError) ? (short) -1 : histogramSlot;
			if (calleeHost == null || calleeHost.length() == 0) {
				calleeHost = callee.getName();
			}
			linkDataMap.addLinkData(caller, callerAgentId, callee, calleeHost, timestamp, slotTime, requestCount);
		}

		return linkDataMap;
	}

	private long getValueToLong(Cell cell) {
		return Bytes.toLong(cell.getValueArray(), cell.getValueOffset());
	}

	private ApplicationBean readCalleeApplication(Buffer buffer) {
		short calleeServiceType = buffer.readShort();
		String calleeApplicationName = buffer.readPrefixedString();
		return new ApplicationBean(calleeApplicationName, calleeServiceType);
	}

	private ApplicationBean readCallerApplication(Buffer row) {
		String callerApplicationName = row.read2PrefixedString();
		short callerServiceType = row.readShort();
		return new ApplicationBean(callerApplicationName, callerServiceType);
	}

	private byte[] getOriginalKey(byte[] rowKey) {
		return rowKeyDistributorByHashPrefix.getOriginalKey(rowKey);
	}
}
